package com.shirley;

import reactor.core.publisher.Flux;

import java.time.Duration;

public class Main {
    public static void main(String[] args) throws InterruptedException {

        Flux<String> flux1 = Flux.just("first", "secode", "third");
        Flux<String> flux2 = Flux.just("1", "2", "3");
        Flux flux3 = Flux.interval(Duration.ofMillis(100));

//        Flux.combineLatest(flux1, flux2, (a, b) -> {return a + ":" + b;})
//                .subscribe(i -> System.out.println(i));

//        Flux.merge(flux3, flux2, flux1)
//                .subscribe(i -> System.out.println(i));
//
        Flux.concat(flux3, flux1, flux2)
                .subscribe(i -> System.out.println(i));

//        Flux.firstWithSignal(flux1, flux2)
//                .subscribe(i -> System.out.println(i));

//
//// 1. 每200ms产生一个自增数(Long)
//        Flux flux = Flux.interval(Duration.ofMillis(200))
//                // 2. 对数据做映射。大于3抛异常
//                .map(input -> {
//                    if (input < 3) return "tick " + input;
//                    throw new RuntimeException("boom");
//                })
//                // 遇到错误重新运行一遍
//                .retry(1)
//                // 将值和运行间隔关联起来（元素转为Tuple2<Long, T>类型）,第一个字段为运行间隔
//                .elapsed();

        // 输出映射后的数据，打印异常
//        flux.subscribe(System.out::println, System.err::println);
//        Thread.sleep(2100);


        //        Flux.just("foo")
//                .map(s -> { throw new IllegalArgumentException(s); })
//                .subscribe(v -> System.out.println("GOT VALUE"),
//                        e -> System.out.println("ERROR: " + e));
//
//        Flux flux = Flux.interval(Duration.ofSeconds(2))
//                .map(i -> {
//                    if ((i %3)==0) {
//                        throw new IllegalArgumentException("Shirley's exception.");
//                    } else {
//                        return i;
//                    }
//                })
//                .doFirst(() -> System.out.println("Begin...."))
//                .doOnError(e -> System.out.println("Get an error: " + e.getMessage()))
//                .doOnNext(i -> System.out.println("OnNext " + i))
////                .doOnRequest(i -> System.out.println("OnRequest " + i))
////                .doOnEach(i -> System.out.println("OnEach " + i))
//                .doOnComplete(() -> System.out.println("Complete"))
////                .doOnTerminate(() -> System.out.println("Terminate"))
//                .take(10L);
//
//        flux.subscribe(i -> {
//            System.out.println("Handle " + i);
//        });
////
//        Thread.sleep(1000000L);
//

        //
//        Flux.just(1,2,3,4,5,6,7)
//                .subscribe(i -> System.out.println("Subscribe: " + i));


        //
//        Flux.just(1, 2, 0)
//                .map(i -> "100 / " + i + " = " + (100 / i)) //this triggers an error with 0
//                .onErrorReturn("Divided by zero :(")
//                .subscribe(a -> System.out.println(a));

        //        Flux.interval(Duration.ofSeconds(2))
//                .onErrorStop()
//        .doOnNext(i -> System.out.println("doOnNext" + i))
//        .subscribe(new BaseSubscriber<Long>() {
//            public void hookOnSubscribe(Subscription subscription) {
//                System.out.println("Subscribed");
//                request(1);
//            }
//            @Override
//            protected void hookOnNext(Long value) {
//                System.out.println(value);
//                request(1);
//            }
//        });
//        Thread.sleep(100000000L);
//
//        Flux<String> flux = Flux.generate(
//                () -> 0,
//                (state, sink) -> {
//                    sink.next("3 x " + state + " = " + 3*state);
//                    if (state == 10) sink.complete();
//                    return state + 1;
//                });
//        flux.subscribe(i -> System.out.println(i));
//
//        Flux<String> bridge = Flux.just("a")
//                .subscribeOn(Schedulers.boundedElastic());


        Thread.sleep(100000L);

    }
}